package com.simple.mq.core.network.core;

import cn.hutool.core.collection.CollectionUtil;
import com.simple.mq.common.util.KeyUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.*;

/**
 * 项目: simple-mq
 * <p>
 * 功能描述: 队列管理器；
 *
 * @author: WuChengXing
 * @create: 2023-09-04 14:26
 **/
@Slf4j(topic = "QueueManager")
public class QueueManager {

    private static List<MessageQueue> queues;

    private static HashMap<String, MessageQueue> queueMap;

    /**
     * 队列管理器初始化
     *
     * @param names2bindingKeys
     * @throws RuntimeException 如果queueNum和bindingKeys的长度不对应，抛出异常
     */
    public static void init(Map<String, String> names2bindingKeys) throws RuntimeException {
        synchronized (QueueManager.class) {
            if (CollectionUtil.isEmpty(queues)) {
                queues = new ArrayList<>();
            }
            if (CollectionUtil.isEmpty(queueMap)) {
                queueMap = new HashMap<>(4);
            }
            for (Map.Entry<String, String> entry : names2bindingKeys.entrySet()) {
                String queueName = entry.getKey();
                String bindingKey = entry.getValue();
                MessageQueue messageQueue = new MessageQueue(bindingKey, queueName);
                queueMap.put(queueName, messageQueue);
                queues.add(messageQueue);
            }
        }
    }

    /**
     * 供外部访问，是否初始化
     *
     * @return
     */
    public static boolean isInited() {
        return false;
    }

    /**
     * 放入一条消息到消息队列中：
     * 这里会根据绑定键的形式来投递消息，存在MessageQueue中；
     *
     * @param message
     * @param routingKey
     */
    public static void put(Object message, String routingKey) {
        for (MessageQueue queue : queues) {
            String bindingKey = queue.getBindingKey();
            if (KeyUtil.routingKeyCompare(routingKey, bindingKey)) {
                try {
                    queue.put(message);
                } catch (InterruptedException e) {
                    // 忽略打断
                }
            }
        }
    }

    public static MessageQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public static boolean containsQueue(String name) {
        return queueMap.containsKey(name);
    }

    /**
     * 唤醒在某个queue上等待的线程
     *
     * @param queueName 队列名
     */
    public static void signalAll(String queueName) {
        MessageQueue queue = queueMap.get(queueName);
        queue.latch.countDown();
    }

    /**
     * 唤醒在某个queue上等待的线程：
     * 这里就是为了配合WorkerManager里面的“无消费者”的等待，消费者注册上来后，会唤醒之前的工作线程，进行消息消费；
     *
     * @param queueName 队列名
     */
    public static void signal(String queueName) {
        try {
            // 这里是把之前因为没有消费者消费的那部分睡眠的工作线程给唤醒
            for (Thread worker : queueMap.get(queueName).workers) {
                worker.interrupt();
            }
        } catch (Exception e) {
            log.error("队列【{}】暂未找到！请先注册队列！", queueName);
        }

    }

    public static HashMap<String, MessageQueue> getQueueMap() {
        return queueMap;
    }
}
